package com.fanml.xyz;

import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.io.Reader;
import java.io.StringWriter;
import java.io.Writer;
import java.sql.Blob;
import java.sql.Clob;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.ResultSetMetaData;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.ArrayList;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CountDownLatch;

/**
 * @param <T>
 * @ClassName: HandleThreadRunable
 * @Description: 批量抽取业务类
 * @author: ChuanHui.Z
 * @date: 2020年9月9日 下午8:45:39
 */
public class HandleThreadRunable<T extends HandleDataMart> implements Runnable {

    private static final int BUFFER_SIZE = 4096;
    private Connection sourceConnection;
    private String sql;
    private CountDownLatch countDownLatch;
    private T object;

    public T getObject() {
        return object;
    }

    public void setObject(T object) {
        this.object = object;
    }

    /**
     * @param sql              分页sql语句
     * @param sourceConnection 源数据源
     * @param countDownLatch   线程阻塞
     * @Title:HandleThreadRunable
     * @Description:TODO 批量获取源数据
     */
    public HandleThreadRunable(String sql, Connection sourceConnection, CountDownLatch countDownLatch) {
        this.sql = sql;
        this.sourceConnection = sourceConnection;
        this.countDownLatch = countDownLatch;
    }

    /*
     * (non Javadoc)
     *
     * @Title: run
     *
     * @Description: 批量获取数据
     *
     * @see java.lang.Runnable#run()
     */
    @Override
    public void run() {
        try {
            List<Map<String, Object>> list = executeQuery(sourceConnection, sql);
            if (list != null && list.size() > 0) {
                object.setListData(list);// 设置源数据集
                object.run();// 执行业务逻辑
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            countDownLatch.countDown();
        }
    }

    /**
     * @param conn
     * @param sql
     * @return
     * @throws SQLException
     * @Title: executeQuery
     * @Description: TODO
     * @return: List<Map < String, Object>>
     */
    public static List<Map<String, Object>> executeQuery(Connection conn, String sql) throws SQLException {
        List<Map<String, Object>> rows = new ArrayList<Map<String, Object>>();
        PreparedStatement stmt = null;
        ResultSet rs = null;
        try {
            stmt = conn.prepareStatement(sql);
            rs = stmt.executeQuery();
            ResultSetMetaData rsMeta = rs.getMetaData();
            while (rs.next()) {
                Map<String, Object> row = new LinkedHashMap<String, Object>();
                for (int i = 0, size = rsMeta.getColumnCount(); i < size; ++i) {
                    int columnType = rsMeta.getColumnType(i + 1);
                    String columName = rsMeta.getColumnLabel(i + 1);
                    if ("RN".equals(columName))
                        continue;
                    if (2004 == columnType) {
                        // ORACLE BLOB
                        Blob blob = rs.getBlob(i + 1);
                        byte[] bs = null;
                        if (blob != null) {
                            InputStream inputStream = blob.getBinaryStream();
                            try {
                                bs = copyToByteArray(inputStream);
                            } catch (IOException e) {
                                e.printStackTrace();
                            }
                        }
                        row.put(columName, bs);
                    } else if (2005 == columnType) {
                        // ORACLE CLOB
                        Clob clob = rs.getClob(i + 1);
                        if (clob != null) {
                            Reader reader = clob.getCharacterStream();
                            String clobString = null;
                            try {
                                clobString = copyToString(reader);
                            } catch (IOException e) {
                                // TODO Auto-generated catch block
                                e.printStackTrace();
                            }
                            row.put(columName, clobString);
                        } else {
                            row.put(columName, null);
                        }
                    } else {
                        Object value = rs.getObject(i + 1);
                        row.put(columName, value);
                    }
                }
                rows.add(row);
            }
        } finally {
            close(rs);
            close(stmt);
//			close(conn);
        }
        return rows;
    }

    public static void close(ResultSet x) {
        if (x == null) {
            return;
        }
        try {
            x.close();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    public static void close(Statement x) {
        if (x == null) {
            return;
        }
        try {
            x.close();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    public static void close(Connection x) {
        if (x == null) {
            return;
        }
        try {
            x.close();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    public static String copyToString(Reader in) throws IOException {
        if (in == null) {
            return "";
        }

        StringWriter out = new StringWriter();
        copy(in, out);
        return out.toString();
    }

    /**
     * @param in  the Reader to copy from
     * @param out the Writer to copy to
     * @return
     * @throws IOException in case of I/O errors
     * @Title: copy
     * @Description: Copy the contents of the given Reader to the given Writer.
     * Closes both when done.
     * @return: int the number of characters copied
     */
    public static int copy(Reader in, Writer out) throws IOException {
        try {
            int byteCount = 0;
            char[] buffer = new char[BUFFER_SIZE];
            int bytesRead = -1;
            while ((bytesRead = in.read(buffer)) != -1) {
                out.write(buffer, 0, bytesRead);
                byteCount += bytesRead;
            }
            out.flush();
            return byteCount;
        } finally {
            try {
                in.close();
            } catch (IOException ex) {
            }
            try {
                out.close();
            } catch (IOException ex) {
            }
        }
    }

    public static byte[] copyToByteArray(InputStream in) throws IOException {
        if (in == null) {
            return new byte[0];
        }

        ByteArrayOutputStream out = new ByteArrayOutputStream(BUFFER_SIZE);
        copy(in, out);
        return out.toByteArray();
    }

    public static int copy(InputStream in, OutputStream out) throws IOException {

        try {
            return copy2(in, out);
        } finally {
            try {
                in.close();
            } catch (IOException ex) {
            }
            try {
                out.close();
            } catch (IOException ex) {
            }
        }
    }

    public static int copy2(InputStream in, OutputStream out) throws IOException {

        int byteCount = 0;
        byte[] buffer = new byte[BUFFER_SIZE];
        int bytesRead = -1;
        while ((bytesRead = in.read(buffer)) != -1) {
            out.write(buffer, 0, bytesRead);
            byteCount += bytesRead;
        }
        out.flush();
        return byteCount;
    }
}
